package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;

import java.time.Duration;

@EnableBinding(value = HotArticleListener.HotArticleProcessor.class)
public class HotArticleListener {
    //监听主题，发送结果主题
    @StreamListener("hot_article_score_topic")
    @SendTo("hot_article_result_topic")
    public KStream<String, String> process(KStream<String, String> input) {
        //从流中获取数据,转换

        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Long articleId = updateArticleMessage.getArticleId();
                return new KeyValue<>(articleId.toString(), value);
            }
        });
        //根据Key分组
        KGroupedStream<String, String> groupByKey = map.groupByKey(Grouped.with(Serdes.String(), Serdes.String()));
        //根据时间窗口来统计
        TimeWindowedKStream<String, String> windowedKStream = groupByKey.windowedBy(TimeWindows.of(Duration.ofSeconds(10)));
        //进行聚合运算
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                return null;
            }
        };
        Aggregator<String,String,String> agg = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                //key指分组的Key，value最开始传递进来的对象，aggregate上一次运算的结果
                long articleId = Long.parseLong(key);
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                ArticleStreamMessage result = null;
                if (StringUtils.isEmpty(aggregate)){
                    result = new ArticleStreamMessage();
                    result.setArticleId(articleId);
                    switch (updateArticleMessage.getType()){
                        case 0:
                            result.setView(updateArticleMessage.getAdd());
                            break;
                        case 1:
                            result.setLike(updateArticleMessage.getAdd());
                            break;
                        case 2:
                            result.setComment(updateArticleMessage.getAdd());
                            break;
                        case 3:
                            result.setCollect(updateArticleMessage.getAdd());
                            break;
                    }
                }else {
                    //从上一次结果转换对象
                    result = JSON.parseObject(aggregate,ArticleStreamMessage.class);
                    switch (updateArticleMessage.getType()){
                        case 0:
                            result.setView(result.getView()+updateArticleMessage.getAdd());
                            break;
                        case 1:
                            result.setLike(result.getLike()+updateArticleMessage.getAdd());
                            break;
                        case 2:
                            result.setComment(result.getComment()+updateArticleMessage.getAdd());
                            break;
                        case 3:
                            result.setCollect(result.getCollect()+updateArticleMessage.getAdd());
                            break;
                    }
                }
                return JSON.toJSONString(result);
            }
        };
        KTable<Windowed<String>, String> aggregate = windowedKStream.aggregate(init, agg, Materialized.with(Serdes.String(), Serdes.String()));
        KStream<String, String> map1 = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return map1;
    }

    public interface HotArticleProcessor {
        @Input("hot_article_score_topic")
        KStream<String, String> input();

        @Output("hot_article_result_topic")
        KStream<String, String> output();
    }
}
